package com.mjf.spark.day11

import java.net.URI

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
 * 优雅的关闭
 *    思路：使用标记位
 */
object SparkStreaming06_GracefulStop {
  def main(args: Array[String]): Unit = {

    // 创建配置文件对象   注意：Streaming程序执行至少需要2个线程（1个用来采集，1一个用来计算），所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")

    // 设置优雅的关闭
    conf.set("spark.streaming.stopGracefullyOnShutdown", "true")

    // 创建SparkStreaming程序执行入口对象   Seconds:时间间隔
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 从指定的端口获取数据
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop103", 9999)

    // 将DS转换为RDD进行操作    DS本身是没有提供排序函数的
    val resDS: DStream[(String, Int)] = socketDS.transform(
      rdd => {
        val flatMapRDD: RDD[String] = rdd.flatMap(_.split(" ")) // 这里是对DStream底层封装的RDD进行操作
        val mapRDD: RDD[(String, Int)] = flatMapRDD.map((_, 1))
        val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
        reduceRDD.sortByKey()
      }
    )

    // 启动新的线程，希望在特殊的场合关闭SparkStreaming
    new Thread(new Runnable {
      override def run(): Unit = {

        while ( true ) {
          try {
            Thread.sleep(5000)
          } catch {
            case ex : Exception => println(ex)
          }

          // 监控HDFS文件的变化
          val fs: FileSystem = FileSystem.get(new URI("hdfs://hadoop102:9000"), new Configuration(), "mjf")

          val state: StreamingContextState = ssc.getState()
          // 如果环境对象处于活动状态，可以进行关闭操作
          if ( state == StreamingContextState.ACTIVE ) {

            // 判断路径是否存在
            val flg: Boolean = fs.exists(new Path("hdfs://hadoop102:9000/stopSpark"))
            if ( flg ) {
              ssc.stop(true, true)
              System.exit(0)
            }

          }
        }

      }
    }).start()

    resDS.print()

    // 启动采集器
    ssc.start()

    // 等待采集结束之后，终止程序
    ssc.awaitTermination()

  }
}
